﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using NUnit.Framework;
using XcoAppSpaces.Contracts.Transport;
using XcoAppSpaces.Serialization.Binary;
using XcoAppSpaces.Transport.Sockets;
using XcoAppSpaces.Contracts.Exceptions;

namespace test.XcoAppSpaces.Transport.Sockets
{
	[TestFixture]
	public class TransferTests
	{
		[Test]
		public void TestTransfer()
		{
			//XcoLoggerFactory.CreateInstance = (t => new TestLogger.TestLogger());
			string addr1 = Environment.MachineName.ToLower() + ":8000";
			string addr2 = Environment.MachineName.ToLower() + ":8001";

			var serializer = new XcoBinarySerializer();
			var msg = new IntMsg {value = 123};
			var receiver1 = new ReceivedTmp();
			var receiver2 = new ReceivedTmp();

			using (XcoTCPTransportService comm1 = new XcoTCPTransportService(8000),
			                              comm2 = new XcoTCPTransportService(8001))
			{
				comm1.OnMessageReceived += receiver1.MessageReceived;
				comm2.OnMessageReceived += receiver2.MessageReceived;

				comm1.Start();
				comm2.Start();

				comm1.Send(new XcoMessage {Content = serializer.Serialize(msg), CausalityContext = null}, addr2);
				comm2.Send(new XcoMessage {Content = serializer.Serialize(msg), CausalityContext = null}, addr1);
				comm1.Send(new XcoMessage {Content = serializer.Serialize(msg), CausalityContext = null}, addr2);

				comm2.Stop();
				Thread.Sleep(500);

				try
				{
					var xcoMsg = new XcoMessage {Content = serializer.Serialize(msg), CausalityContext = null};
					Exception expectedEx = null;
					xcoMsg.OnTransferFailed += (_, ex) => expectedEx = ex;
					comm1.Send(xcoMsg, addr2);
					Assert.IsNotNull(expectedEx, "Should not be able to send message!");
				}
				catch (XcoCommunicationException)
				{
				}
				Thread.Sleep(1000);
				//Assert.AreEqual(1, receiver1.Counter);
				Assert.AreEqual(2, receiver2.Counter);
			}

			using (XcoTCPTransportService comm1 = new XcoTCPTransportService(8000),
			                              comm2 = new XcoTCPTransportService(8001))
			{
				comm1.OnMessageReceived += receiver1.MessageReceived;
				comm2.OnMessageReceived += receiver2.MessageReceived;
				comm1.Start();
				comm2.Start();

				comm1.Send(new XcoMessage {Content = serializer.Serialize(msg), CausalityContext = null}, addr2);
				comm2.Send(new XcoMessage {Content = serializer.Serialize(msg), CausalityContext = null}, addr1);
				comm1.Send(new XcoMessage {Content = serializer.Serialize(msg), CausalityContext = null}, addr2);

				comm2.Stop();
				Thread.Sleep(500);

				try
				{
					var xcoMsg = new XcoMessage {Content = serializer.Serialize(msg), CausalityContext = null};
					Exception expectedEx = null;
					xcoMsg.OnTransferFailed += (_, ex) => expectedEx = ex;
					comm1.Send(xcoMsg, addr2);
					Assert.IsNotNull(expectedEx, "Should not be able to send message!");
				}
				catch (XcoCommunicationException)
				{
				}
				Thread.Sleep(1000);
				//Assert.AreEqual(2, receiver1.Counter);
				Assert.AreEqual(4, receiver2.Counter);
			}
		}

		[Test]
		public void TestActiveConnections()
		{
			var serializer = new XcoBinarySerializer();
			var msg = new IntMsg {value = 123};
			var receiver1 = new ReceivedTmp();
			var receiver2 = new ReceivedTmp();

			using (XcoTCPTransportService comm1 = new XcoTCPTransportService(0),
			                              comm2 = new XcoTCPTransportService(0))
			{
				Assert.IsNull(comm1.ActiveConnections);
				Assert.IsNull(comm2.ActiveConnections);

				comm1.OnMessageReceived += receiver1.MessageReceived;
				comm2.OnMessageReceived += receiver2.MessageReceived;
				comm1.Start();
				comm2.Start();
				var addr1 = comm1.LocalAddress;
				var addr2 = comm2.LocalAddress;

				Assert.AreEqual(0, comm1.ActiveConnections.Count());
				Assert.AreEqual(0, comm2.ActiveConnections.Count());

				comm1.Send(new XcoMessage {Content = serializer.Serialize(msg), CausalityContext = null}, addr2);
				Thread.Sleep(100);
				Assert.AreEqual(1, comm1.ActiveConnections.Count());
				Assert.AreEqual(1, comm2.ActiveConnections.Count());

				comm2.Send(new XcoMessage {Content = serializer.Serialize(msg), CausalityContext = null}, addr1);
				Thread.Sleep(100);
				Assert.AreEqual(1, comm1.ActiveConnections.Count());
				Assert.AreEqual(1, comm2.ActiveConnections.Count());


				Assert.AreEqual(addr2, comm1.ActiveConnections.First());
				Assert.AreEqual(addr1, comm2.ActiveConnections.First());
			}
		}

		[Test]
		public void ConcurrentTransfer()
		{
			string addr2 = Environment.MachineName.ToLower() + ":8001";

			using (XcoTCPTransportService comm1 = new XcoTCPTransportService(8000),
			                              comm2 = new XcoTCPTransportService(8001))
			{
				int received = 0;
				comm2.OnMessageReceived += (message, address, service) => Interlocked.Increment(ref received);

				comm1.Start();
				comm2.Start();

				const int sent = 20;
				int successFullyTransferred = 0;
				int transferErrors = 0;
				for (int i = 0; i < sent; i++)
				{
					System.Threading.Tasks.Task.Factory.StartNew(() =>
						{
							var msg = new XcoMessage {Content = new byte[] {1, 2, 3}, CausalityContext = null};
							msg.OnTransferSuccessful += _ => Interlocked.Increment(ref successFullyTransferred);
							msg.OnTransferFailed += (m,ex) => Interlocked.Increment(ref transferErrors);
							comm1.Send(msg, addr2);
						});
				}

				Assert.IsTrue(SpinWait.SpinUntil(() => received == sent, 1000));
				Assert.AreEqual(sent, successFullyTransferred);
				Assert.AreEqual(0, transferErrors);
			}
		}

		[Test]
		public void ConcurrentFailedTransfer()
		{
			string addr2 = Environment.MachineName.ToLower() + ":8001";

			using (var comm1 = new XcoTCPTransportService(8000))
			{
				comm1.Start();

				const int sent = 20;
				int successFullyTransferred = 0;
				int transferErrors = 0;
				for (int i = 0; i < sent; i++)
				{
					System.Threading.Tasks.Task.Factory.StartNew(() =>
					{
						var msg = new XcoMessage { Content = new byte[] { 1, 2, 3 }, CausalityContext = null };
						msg.OnTransferSuccessful += _ => Interlocked.Increment(ref successFullyTransferred);
						msg.OnTransferFailed += (m, ex) => Interlocked.Increment(ref transferErrors);
						comm1.Send(msg, addr2);
					});
				}

				SpinWait.SpinUntil(() => transferErrors == sent, 2000);
				Assert.AreEqual(sent, transferErrors);
				Assert.AreEqual(0, successFullyTransferred);
			}
		}

		[Test]
		public void Failed_Transfer_after_Successful_Transfer()
		{
			string addr2 = Environment.MachineName.ToLower() + ":8001";

			using (XcoTCPTransportService comm1 = new XcoTCPTransportService(8000),
										  comm2 = new XcoTCPTransportService(8001))
			{
				int received = 0;
				comm2.OnMessageReceived += (message, address, service) => Interlocked.Increment(ref received);

				comm1.Start();
				comm2.Start();

				int successFullyTransferred = 0;
				int transferErrors = 0;
				var msg = new XcoMessage { Content = new byte[] { 1, 2, 3 }, CausalityContext = null };
				msg.OnTransferSuccessful += _ => Interlocked.Increment(ref successFullyTransferred);
				msg.OnTransferFailed += (m, ex) => { Interlocked.Increment(ref transferErrors); Console.WriteLine(ex.Message); };

				//successful send
				comm1.Send(msg, addr2);
				Thread.Sleep(200);

				comm2.Dispose();
				
				//failed send
				comm1.Send(msg, addr2);
				Thread.Sleep(200);
				
				Assert.AreEqual(1, successFullyTransferred);
				Assert.AreEqual(1, transferErrors);
			}
		}

		[Test, Explicit]
		[Repeat(100)]
		public void TestConcurrentDualConnectionCreation()
		{
			//Check if by opening the connection at the same time from both sides a state could be triggered where a message doesn't arrive

			string addr1 = Environment.MachineName.ToLower() + ":8000";
			string addr2 = Environment.MachineName.ToLower() + ":8001";

			var serializer = new XcoBinarySerializer();
			var msg = new byte[] {1, 2, 3};
			var count1 = 0;
			var count2 = 0;
		
			using (XcoTCPTransportService comm1 = new XcoTCPTransportService(8000),
			                              comm2 = new XcoTCPTransportService(8001))
			{
				comm1.OnMessageReceived += (message, address, service) =>
					{
						//Console.WriteLine("Comm 1 message received.");
						Interlocked.Increment(ref count1);
					};
				comm2.OnMessageReceived += (message, address, service) =>
					{
						//Console.WriteLine("Comm 2 message received.");
						Interlocked.Increment(ref count2);
					};
				
				comm1.Start();
				comm2.Start();

				var barrier = new Barrier(2);
				var thread1 = new Thread(() =>
					{
						barrier.SignalAndWait();
						//Console.WriteLine("comm1 start sending");
						for (int i = 0; i < 5; i++)
						{
							comm1.Send(new XcoMessage { Content = msg, CausalityContext = null }, addr2);
							Thread.Sleep(100);
						}
					});
				var thread2 = new Thread(() =>
					{
						barrier.SignalAndWait();
						//Console.WriteLine("comm2 start sending");
						for (int i = 0; i < 5; i++)
						{
							comm2.Send(new XcoMessage { Content = msg, CausalityContext = null }, addr1);
							Thread.Sleep(100);
						}
					});
				thread1.Start();
				thread2.Start();

				SpinWait.SpinUntil(() => count1 == 5 && count2 == 5, 2000);
				Assert.AreEqual(5, count1);
				Assert.AreEqual(5, count2);
			}
		}
	}
}
